package pers.amos.eshop.storm.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.concurrent.CountDownLatch;

/**
 * @author amos wong
 * @create 2020-11-22 10:05 上午
 */
@Slf4j
public class ZookeeperSession {

    private static final CountDownLatch countDownLatch = new CountDownLatch(1);

    private ZooKeeper zooKeeper = null;

    public ZookeeperSession() {
        try {
            zooKeeper = new ZooKeeper("127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183",
                    50000,
                    new ZookeeperWatcher());
            log.info("connect state:{}", zooKeeper.getState());
            countDownLatch.await();
            log.info("zookeeper connect success!");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * zk是异步连接的，该类为监视器 用于监听zk的连接状态
     */
    private class ZookeeperWatcher implements Watcher {
        @Override
        public void process(WatchedEvent watchedEvent) {
            log.info("received connect state:{}", watchedEvent.getState());
            if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                countDownLatch.countDown();
            }
        }
    }

    /**
     * 获取分布式锁
     */
    public void acquireDistributeLock(String path) {
        try {
            zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL);
        } catch (Exception ex) {
            int count = 0;
            while (true) {
                try {
                    Thread.sleep(20);
                    zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                            CreateMode.EPHEMERAL);
                } catch (Exception e) {
                    count++;
                    log.info("try acquire lock {} times", count);
                    continue;
                }
                log.info("acquire distribute lock success!");
                break;
            }
        }
    }

    /**
     * 释放分布式锁
     */
    public void releaseDistributeLock(String path) {
        try {
            zooKeeper.delete(path, -1);
            log.info("release distribute lock success! path={}", path);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取任务列表节点的值
     * 即获取所有storm的任务节点
     *
     * @return
     */
    public String getNodeData(String path) {
        try {
            return new String(zooKeeper.getData(path, false, new Stat()));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "";
    }

    /**
     * 设置节点的值
     *
     * @param path
     * @param data
     */
    public void setNodeData(String path, String data) {
        try {
            zooKeeper.setData(path, data.getBytes(), -1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建节点，当节点不存在时才会去创建
     * @param path
     */
    public void createNode(String path) {
        try {
            zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch (Exception e) {

        }
    }

    private static class Singleton {

        private static ZookeeperSession instance;

        static {
            instance = new ZookeeperSession();
        }

        public static ZookeeperSession getSingleton() {
            return instance;
        }
    }

    public static ZookeeperSession getInstance() {
        return Singleton.getSingleton();
    }

    public static void init() {
        getInstance();
    }

}
